package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.VisitorStats;
import com.atguigu.gmall.realtime.utils.DateTimeUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;

/**
 * Desc: 访客主题宽表计算
 * <p>
 * ?要不要把多个明细的同样的维度统计在一起?
 * 因为单位时间内mid的操作数据非常有限不能明显的压缩数据量（如果是数据量够大，或者单位时间够长可以）
 * 所以用常用统计的四个维度进行聚合 渠道、新老用户、app版本、省市区域
 * 度量值包括 启动、日活（当日首次启动）、访问页面数、新增用户数、跳出数、平均页面停留时长、总访问时长
 * 聚合窗口： 10秒
 * <p>
 * 各个数据在维度聚合前不具备关联性，所以先进行维度聚合
 * 进行关联  这是一个fulljoin
 * 可以考虑使用flinksql 完成
 */

//数据流 web/app -> nginx -> SpringBoot -> Kafka(ods) -> FlinkApp -> Kafka(dwd) -> FlinkApp -> Kafka(dwm)
// ->  FlinkApp -> ClickHouse
//程序 mocklog -> Nginx -> Logger.sh -> Kafka(zk) -> BaseLogApp -> Kafka -> uv/uj -> Kafka
// -> VisitorStatsApp -> ClickHouse

public class VisitorStatsApp {
    public static void main(String[] args) throws Exception {
//        1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//        //2.Flink-CDC将读取binlog的位置信息以状态的方式保存在CK,如果想要做到断点续传,需要从Checkpoint或者Savepoint启动程序
//        //2.1 开启Checkpoint,每隔5秒钟做一次CK
//        env.enableCheckpointing(5000L);
////        设置超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(10000L);
////        设置最大当前ck数量
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
////        设置最小ck间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);
//        //2.2 指定CK的一致性语义
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        //2.3 设置任务关闭的时候保留最后一次CK数据
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup
//        .RETAIN_ON_CANCELLATION);
//        //2.4 指定从CK自动重启策略
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
//        //2.5 设置状态后端
//        env.setStateBackend(new FsStateBackend("hdfs://djcluster/flinkCDC"));
//        //2.6 设置访问HDFS的用户名
//        System.setProperty("HADOOP_USER_NAME", "atguigu");
//        2.读取Kafka数据

        //TODO 1.从Kafka的pv、uv、跳转明细主题中获取数据
        String groupId = "visitor_stats_app";
        String uniqueVisitSourceTopic = "dwm_unique_visit";
        String userJumpDetailSourceTopic = "dwm_user_jump_detail";
        String pageViewSourceTopic = "dwd_page_log";
        DataStreamSource<String> uvDS =
                env.addSource(MyKafkaUtil.getKafkaConsumer(uniqueVisitSourceTopic, groupId));
        DataStreamSource<String> ujDS =
                env.addSource(MyKafkaUtil.getKafkaConsumer(userJumpDetailSourceTopic, groupId));
        DataStreamSource<String> pvDS =
                env.addSource(MyKafkaUtil.getKafkaConsumer(pageViewSourceTopic, groupId));

//        3.将将每个流处理成相同的数据类型
//        3.1处理UV数据
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithUvDS = uvDS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);

//          提取公共字段
            JSONObject common = jsonObject.getJSONObject("common");
            return new VisitorStats("", ""
                    , common.getString("vc")
                    , common.getString("ch")
                    , common.getString("ar")
                    , common.getString("is_new")
                    , 1L
                    , 0L, 0L, 0L, 0L, jsonObject.getLong("ts")
            );
        });
        //        3.2处理UJ数据
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithUjDS = ujDS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            //提取公共字段
            JSONObject common = jsonObject.getJSONObject("common");
            return new VisitorStats("", "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    0L, 0L, 0L, 1L, 0L,
                    jsonObject.getLong("ts"));
        });
        //3.3处理PV数据
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithPvDS = pvDS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
//            获取公共字段
            JSONObject common = jsonObject.getJSONObject("common");
//            获取页面信息
            JSONObject page = jsonObject.getJSONObject("page");
            String last_page_id = page.getString("last_page_id");
            long sv = 0L;

//          上一跳页面是null
            if (last_page_id == null || last_page_id.length() <= 0) {
                sv = 1L;
            }

            return new VisitorStats(
                    "", ""
                    , common.getString("vc")
                    , common.getString("ch")
                    , common.getString("ar")
                    , common.getString("is_new")
                    , 0L
                    , 1L, sv, 0L, page.getLong("during_time")
                    , jsonObject.getLong("ts")
            );
        });


//        4.Union几个流

        DataStream<VisitorStats> unionDS = visitorStatsWithUvDS.union(
                visitorStatsWithUjDS
                , visitorStatsWithPvDS);

//        5.提取时间戳生成WaterMark  开窗 事件时间 就用乱序水位线
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithWMDS =
                unionDS.assignTimestampsAndWatermarks(WatermarkStrategy.<VisitorStats>forBoundedOutOfOrderness(
//                        必须要大于上面Jump那边的延迟时间
                        Duration.ofSeconds(11))
                        .withTimestampAssigner(new SerializableTimestampAssigner<VisitorStats>() {
                            @Override
                            public long extractTimestamp(VisitorStats element, long recordTimestamp) {
                                return element.getTs();
                            }
                        }));

//        6.按照维度信息进行分组 返回值 4元组
        KeyedStream<VisitorStats, Tuple4<String, String, String, String>> keyedStream =
                visitorStatsWithWMDS.keyBy(new KeySelector<VisitorStats, Tuple4<String, String, String, String>>() {
                    @Override
                    public Tuple4<String, String, String, String> getKey(VisitorStats value) throws Exception {
                        return new Tuple4<>(
//                        4元组的Key
                                value.getAr()
                                , value.getCh()
                                , value.getIs_new()
                                , value.getVc());
                    }
                });

//        7.开窗聚合 10s的窗口
        WindowedStream<VisitorStats, Tuple4<String, String, String, String>, TimeWindow> windowedStream =
                keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(10)));
//  增量聚合跟全量聚合一起来使用      reduce + window    来一条聚合一条用reduce   窗口结果的时候 window接管reduce
        SingleOutputStreamOperator<VisitorStats> result = windowedStream.reduce(new ReduceFunction<VisitorStats>() {
            @Override
            public VisitorStats reduce(VisitorStats value1, VisitorStats value2) throws Exception {
//                如果开的是滑动窗口只能是下面这种
//                return new VisitorStats(value1.getStt()
//                        , value1.getEdt()
//                        , value1.getVc()
//                        , value1.getCh())
//                        , value1.getAr()
//                        , value1.getIs_new()
//                        ,value1.getUv_ct()+value2.getUv_ct()
//                        ;
                value1.setUv_ct(value1.getUv_ct() + value2.getUv_ct());
                value1.setPv_ct(value1.getPv_ct() + value2.getPv_ct());
                value1.setSv_ct(value1.getSv_ct() + value2.getSv_ct());
                value1.setUj_ct(value1.getUj_ct() + value2.getUj_ct());
                value1.setDur_sum(value1.getDur_sum() + value2.getDur_sum());

                return value1;
            }
        }, new WindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>() {
            @Override
            public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4,
                              TimeWindow window, Iterable<VisitorStats> input, Collector<VisitorStats> out) throws Exception {

                long start = window.getStart();
                long end = window.getEnd();

//                窗口结束只有一条数据输出
                VisitorStats visitorStats = input.iterator().next();

//                补充窗口信息
                visitorStats.setStt(DateTimeUtil.toYMDhms(new Date(start)));
                visitorStats.setEdt(DateTimeUtil.toYMDhms(new Date(end)));

                out.collect(visitorStats);
            }
        });

//        将数据写入ClickHouse
        result.print(">>");

//        8.启动任务

        env.execute("VisitorStatsApp");
    }

}
